package com.rabbitmq.topics;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

// 消费者
public class Consumer {

    private class ComsumerThread implements Runnable{

        @Override
        public void run() {
            //1. 获取ConnectFactory对象，设置主机、端口、用户名、密码、虚拟主机参数
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("47.122.113.100");
            factory.setPort(5672);
            factory.setUsername("admin");
            factory.setPassword("123");
            factory.setVirtualHost("/");

            //2. 通过ConnectFactory创建Connection对象，创建通道Channel，声明队列
            Connection connection = null;
            Channel channel = null;
            String queueName = Thread.currentThread().getName();


            try {
                connection = factory.newConnection();
                channel = connection.createChannel();
                /**
                 * queue:队列名
                 * durable:是否持久化(不持久化也会存储在磁盘上，但是服务重启后会丢失)
                 * exclusive:是否排他（是否独占一个队列）
                 * autoDelete:是否自动删除（在所有信息被消费之后，是否自动删除队列）
                 * arguments:参数
                 */
                channel.queueDeclare("hello", false, false, false, null);

                //3. 消费消息，持续监听
                channel.basicConsume(queueName, true, new DeliverCallback() {
                    @Override
                    public void handle(String consumerTag, Delivery message) throws IOException {
                        System.out.println(Thread.currentThread().getName()+"接收到的消息："+new String(message.getBody(),"UTF-8"));
                    }
                },new CancelCallback() {
                    @Override
                    public void handle(String consumerTag) throws IOException {
                        System.out.println("消费失败");
                    }
                });

                System.in.read();   //阻塞线程，等待输入



            }catch (Exception e){
                e.printStackTrace();
            }finally {
                //4. 关闭通道，关闭连接
                if (channel != null && channel.isOpen()){
                    try {
                        channel.close();
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    } catch (TimeoutException e) {
                        throw new RuntimeException(e);
                    }
                }

                if (connection != null && connection.isOpen()){
                    try {
                        connection.close();
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }

            }
        }
    }

    public static void main(String[] args) {
        new Thread(new Consumer().new ComsumerThread(),"queue1").start();
        new Thread(new Consumer().new ComsumerThread(),"queue2").start();
        new Thread(new Consumer().new ComsumerThread(),"queue3").start();


    }
}
